import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

/**
 * 读取hdfs文件，在本地运行spark, 控制台打印，
 */

public class WordCountApp {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf()
            .setAppName("Word Count")
                .setMaster("local[*]"); // 使用 local[*] 指示 Spark 使用本地所有可用的处理器
//            .setMaster("local"); // 使用 local 作为 Master URL，适合开发环境  本地运行

        JavaSparkContext sc = new JavaSparkContext(conf);

        // 读取 HDFS 上的数据
        JavaRDD<String> textFile = sc.textFile("hdfs://192.168.1.201:9000//text.log");
//        JavaRDD<String> textFile = sc.textFile("C:\\Users\\govnet\\Desktop\\text.log");

        // 分割单词
        JavaRDD<String> words = textFile.flatMap(s -> Arrays.asList(s.split("\\W+")).iterator());

        // 计数
        JavaPairRDD<String, Integer> wordCounts = words.mapToPair(word -> new Tuple2<>(word, 1))
            .reduceByKey((a, b) -> a + b);

        // 输出结果
        wordCounts.foreach(tuple -> System.out.println(tuple._1() + ": " + tuple._2()));

        sc.close();
    }
}